/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package simple.utils.httpclient;

import com.google.common.base.Preconditions;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.SocketChannel;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.http.DefaultHttpRequest;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
import org.jboss.netty.util.HashedWheelTimer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author rehak
 */
public class HTTPClient implements Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(HTTPClient.class);
    private static final int DEFAULT_MAX_CONTENT_LENGTH = 64 * 1024 * 1024;
    private final LinkedBlockingQueue<DefaultHttpResponse> queue = new LinkedBlockingQueue<>();
    private final ChannelGroup channelGroup = new DefaultChannelGroup("SSLClientChannels");
    private final String host;
    private final int port;
    private Channel channel;
    private final ClientBootstrap bootstrap;
    private final InetSocketAddress socketAddress;
    private final HTTPClientPipelineFactory factory;
    private final NioClientSocketChannelFactory nioFactory;
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final boolean useSSL;
    private final HashedWheelTimer timer;

    public HTTPClient(String host, int port, boolean useSSL) {
        this(host, port, useSSL, DEFAULT_MAX_CONTENT_LENGTH);
    }

    public HTTPClient(String host, int port, boolean useSSL, int maxContentLength) {
        Preconditions.checkNotNull(host, "Host is null!");
        Preconditions.checkArgument(port > 0 && port < 65536, "Port must be > 0 & < 65536");
//        Preconditions.checkArgument(httpReadTimeout>0);
        this.useSSL = useSSL;
        this.socketAddress = new InetSocketAddress(host, port);
        this.host = host;
        this.port = port;
        this.timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS, 1000);
//        this.factory = new HTTPClientPipelineFactory(queue, httpReadTimeout, useSSL, timer);        
        this.factory = new HTTPClientPipelineFactory(queue, useSSL, timer, maxContentLength);
        // Configure the client.
        nioFactory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool()) {

            @Override
            public SocketChannel newChannel(ChannelPipeline pipeline) {
                SocketChannel c = super.newChannel(pipeline);
                channelGroup.add(c);
                return c;
            }
        };
        bootstrap = new ClientBootstrap(nioFactory);
        // Configure the pipeline factory.        
        bootstrap.setPipelineFactory(factory);
        bootstrap.setOption("tcpNoDelay", true);
        bootstrap.setOption("keepAlive", true);
        bootstrap.setOption("remoteAddress", socketAddress);
    }

    public DefaultHttpResponse send(DefaultHttpRequest request, long timeoutMillis) throws Exception {
        connect();
        LOGGER.debug("Sending " + request);
        ChannelFuture writeFuture = channel.write(request);
        writeFuture.awaitUninterruptibly();
        if (!writeFuture.isSuccess()) {
            throw new IOException("Exception while sending request!", writeFuture.getCause());
        }
        LOGGER.debug("Write done, retrieving response");
//        DefaultHttpResponse resp = queue.take();
        DefaultHttpResponse resp = queue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
        LOGGER.debug("Got response " + resp);
        ChannelFuture close = channel.close();
        close.awaitUninterruptibly();
        if (!close.isSuccess()) {
            throw new IOException("Exception while sending request!", close.getCause());
        }
        LOGGER.debug("Send complete, retrieving from queue");
        return resp;
    }

    private void connect() throws IOException {
        Lock readLock = lock.readLock();
        readLock.lock();
        try {
            if (channel == null || !channel.isOpen()) {
                readLock.unlock();
                Lock writeLock = lock.writeLock();
                writeLock.lock();
                try {
                    if (channel == null || !channel.isOpen()) {
                        LOGGER.debug("Connecting to the server");
                        ChannelFuture connectFuture = bootstrap.connect();
                        connectFuture.awaitUninterruptibly();
                        if (!connectFuture.isSuccess()) {
                            throw new IOException("Unable to connect!");
                        }
                        channel = connectFuture.getChannel();
                    }
                } finally {
                    try {
                        readLock.lock();
                    } finally {
                        writeLock.unlock();
                    }
                }
            }
        } finally {
            readLock.unlock();
        }
    }

    @Override
    public void close() throws IOException {
        int size = channelGroup.size();
        try {
            LOGGER.debug("Closing " + size + " channels");
            ChannelGroupFuture future = channelGroup.close();
            future.awaitUninterruptibly();
        } catch (Throwable t) {
            LOGGER.error("Error closing channels", t);
        }
        try {
            LOGGER.debug("Releasing external resources...");
            nioFactory.releaseExternalResources();
        } catch (Throwable t) {
            LOGGER.error("Error releasing external resources", t);
        }
        factory.close();
    }
}
